package com.zhang.flink.exercise;

import com.zhang.flink.bean.OrderEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;

/**
 * @title:
 * @author: zhangyifan
 * @date: 2022/12/16 14:53
 */
public class OrderTimeOut {
    public static void main(String[] args) throws Exception {
        // 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 处理逻辑
        env
                .readTextFile("src/main/resources/OrderLog.csv")
                        .map(new MapFunction<String, OrderEvent>() {
                            @Override
                            public OrderEvent map(String value) throws Exception {
                                String[] fields = value.split(",");
                                return new OrderEvent(
                                        fields[0],fields[1],fields[2],
                                        Long.parseLong(fields[3])*1000L
                                );
                            }
                        })
                                .assignTimestampsAndWatermarks(
                                        WatermarkStrategy.<OrderEvent>forMonotonousTimestamps()
                                                .withTimestampAssigner(new SerializableTimestampAssigner<OrderEvent>() {
                                                    @Override
                                                    public long extractTimestamp(OrderEvent element, long recordTimestamp) {
                                                        return element.ts;
                                                    }
                                                })
                                )
                                        .keyBy(r->r.orderId)
                                                .process(new OrderTimerOutProcess());


        // 提交任务
        env.execute();
    }


    static class OrderTimerOutProcess extends KeyedProcessFunction<String,OrderEvent,String>{
        // 定义状态变量
        private ValueState<OrderEvent> orderEventState;
        private ValueState<Long> timerTs;

        @Override
        public void open(Configuration parameters) throws Exception {
            orderEventState = getRuntimeContext().getState(
                    new ValueStateDescriptor<OrderEvent>(
                            "order-state", TypeInformation.of(OrderEvent.class)
                    )
            );

            timerTs = getRuntimeContext().getState(
                    new ValueStateDescriptor<Long>(
                            "timer",TypeInformation.of(Long.class)
                    )
            );

        }

        @Override
        public void processElement(OrderEvent value, KeyedProcessFunction<String, OrderEvent, String>.Context ctx, Collector<String> out) throws Exception {
            if ("create".equals(value.type)){
                // 如果创建订单超成功，存入状态等待15分钟
                orderEventState.update(value);
                // 注册15分钟后的定时器
                long ts = value.ts + 15 * 60 * 1000L;
                ctx.timerService().registerEventTimeTimer(ts);
                // 保存定时器时间
                timerTs.update(ts);
            }else if ("pay".equals(value.type)){
                OrderEvent orderEvent = orderEventState.value();
                if (orderEvent==null){
                    // 由于网络传输，创建订单数据丢失，只有支付数据
                }else {
                    // 正常支付
                    ctx.timerService().deleteEventTimeTimer(timerTs.value());
                    orderEventState.clear();
                    timerTs.clear();
                }
            }

        }
    }
}
